package org.apache.ocean;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;

import org.apache.commons.lang.StringUtils;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.Term;
import org.apache.lucene.index.TermEnum;
import org.apache.lucene.search.HitCollector;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.Weight;
import org.apache.ocean.cache.Cache;
import org.apache.ocean.server.DateFieldType;
import org.apache.ocean.server.DoubleFieldType;
import org.apache.ocean.server.FieldType;
import org.apache.ocean.server.LocalizedFieldType;
import org.apache.ocean.server.LongFieldType;
import org.apache.ocean.server.StringFieldType;
import org.apache.ocean.util.BitDocSet;
import org.apache.ocean.util.DistanceUtils;
import org.apache.ocean.util.DocSet;

import com.ibm.icu.text.CollationKey;
import com.ibm.icu.util.ULocale;

/**
 * returns scoredocs
 * 
 * @author jasonr
 */
// TODO: unit test ShardSearch
public class ShardSearch {
	public static Logger log = Logger.getLogger(ShardSearch.class.getName());
	SearchIndexShard searchIndexShard;
	long offset;
	long length;
	long startTime;
	int maxDoc;
	int[] starts;
	List<SegmentIndexSearcher> indexSearchers;
	TimeoutState timeoutState;
	IndexSchema indexSchema;
	List<Query> filterQueries;
	FacetResponse facetResponse;
	Cache cache;

	public ShardSearch(long offset, long length, List<Query> filterQueries, FacetResponse facetResponse,
			SearchIndexShard searchIndexShard, TimeoutState timeoutState, IndexSchema indexSchema) throws Exception {
		this.offset = offset;
		this.length = length;
		this.filterQueries = filterQueries;
		this.facetResponse = facetResponse;
		this.searchIndexShard = searchIndexShard;
		this.indexSchema = indexSchema;
		startTime = System.currentTimeMillis();
		indexSearchers = searchIndexShard.createSegmentIndexSearchers();
		this.timeoutState = timeoutState;
		cache = searchIndexShard.getCache();
		starts = new int[indexSearchers.size() + 1]; // build starts array
		int i = 0;
		maxDoc = 0;
		for (IndexSearcher indexSearcher : indexSearchers) {
			starts[i] = maxDoc;
			maxDoc += indexSearcher.maxDoc();
			i++;
		}
		starts[indexSearchers.size()] = maxDoc;
	}

	public int maxDoc() {
		return maxDoc;
	}
  /**
	public int getUniqueTermCount(String field) throws Exception {
		Set<String> termSet = new HashSet<String>();
		List<MultiIndexSearcher> indexSearchers = searchIndexShard.createMultiIndexSearchers();
		for (MultiIndexSearcher indexSearcher : indexSearchers) {
			TermEnum termEnum = indexSearcher.getIndexReader().terms(new Term(field, ""));
			try {
				while (termEnum.next()) {
					Term term = termEnum.term();
					termSet.add(term.text());
				}
			} finally {
				if (termEnum != null)
					termEnum.close();
			}
		}
		return termSet.size();
	}
   **/
	public CDouble createDistanceCurved(CDouble latitude, CDouble longitude, Distance distance) {
		return new CDouble(DistanceUtils.getDistanceMi(distance.latitude.value, distance.longitude.value, latitude.value, longitude.value));
	}

	public SearchResult search(Weight weight, CSortFields sortFields, Distance distance) throws Exception {
		long numDocsReturned = offset + length;
		FieldReaderSortedQueue queue = new FieldReaderSortedQueue(sortFields, numDocsReturned);
		int x = 0;
		CLong numFound = new CLong(0);
		for (SegmentIndexSearcher indexSearcher : indexSearchers) {
			IndexReader indexReader = indexSearcher.getIndexReader();
			DocSet filterDocSet = null;
			if (filterQueries != null && filterQueries.size() > 0) {
				filterDocSet = cache.getDocSet(filterQueries, timeoutState, indexReader);
			}
			FieldHitCollector hitCollector = new FieldHitCollector(filterDocSet, sortFields, queue, indexSearcher, starts[x], distance);
			indexSearcher.search(weight, null, hitCollector);
			processFacets(hitCollector.docSet, indexSearcher);
			numFound.value += hitCollector.numHits;
			x++;
		}
		List<CScoreDoc> scoreDocs = ClusterUtil.toScoreDocs((int) offset, (int) length, queue);
		return new SearchResult(scoreDocs, numFound, indexSchema);
	}

	public class FieldHitCollector extends HitCollector {
		DocSet filter;
		long numHits = 0;
		long lastDocRequested;
		int globalDocStart;
		IndexSearcher indexSearcher;
		IndexReader indexReader;
		CLong[] idArray;
		CLong[] versionArray;
		CSortFields sortFields;
		FieldReaderSortedQueue queue;
		List<Comparable[]> comparableSortArrays;
		CDouble[] latitudeArray;
		CDouble[] longitudeArray;
		Distance distance;
		int distanceSortIndex = -1;
		public DocSet docSet;

		public FieldHitCollector(DocSet filter, CSortFields sortFields, FieldReaderSortedQueue queue, IndexSearcher indexSearcher, int globalDocStart,
				Distance distance) throws Exception {
			this.filter = filter;
			this.sortFields = sortFields;
			this.queue = queue;
			this.indexSearcher = indexSearcher;
			this.indexReader = indexSearcher.getIndexReader();
			if (facetResponse != null)
				docSet = new BitDocSet(indexSearcher.maxDoc());
			lastDocRequested = offset + length;
			this.globalDocStart = globalDocStart;
			this.distance = distance;
			IndexReader indexReader = indexSearcher.getIndexReader();
			idArray = cache.getLongArray(ClusterConstants.ID_FIELD, indexReader);
			versionArray = cache.getLongArray(ClusterConstants.VERSION_FIELD, indexReader);
			if (distance != null) {
				latitudeArray = cache.getDoubleArray(ClusterConstants.LATITUDE_FIELD, indexReader);
				longitudeArray = cache.getDoubleArray(ClusterConstants.LONGITUDE_FIELD, indexReader);
			}
			comparableSortArrays = new ArrayList<Comparable[]>(sortFields.size());
			int sortIndex = 0;

			for (CSortField sortField : sortFields) {
				if (StringUtils.equals(CSortField.DISTANCE, sortField.field)) {
					distanceSortIndex = sortIndex;
				} else {
					// TODO: put this in a method and add type double
					// TODO: throw error if field does not exist
					// TODO: throw error if field is not sortable
					FieldType fieldType = indexSchema.getField(sortField.field).getFieldType();
					if (fieldType instanceof DateFieldType || fieldType instanceof LongFieldType) {
						CLong[] longArray = cache.getLongArray(sortField.field, indexReader);
						comparableSortArrays.add(longArray);
					} else if (fieldType instanceof LocalizedFieldType) {
						LocalizedFieldType localizedFieldType = (LocalizedFieldType) fieldType;
						ULocale locale = localizedFieldType.getLocale();
						CollationKey[] collationKeys = cache.getCollationKeyArray(sortField.field, locale, indexReader);
						comparableSortArrays.add(collationKeys);
					} else if (fieldType instanceof DoubleFieldType) {
						CDouble[] doubleArray = cache.getDoubleArray(sortField.field, indexReader);
						comparableSortArrays.add(doubleArray);
					} else if (fieldType instanceof StringFieldType) {
						StringFieldType stringFieldType = (StringFieldType) fieldType;
						ULocale locale = stringFieldType.getLocale();
						if (locale == null)
							locale = ClusterUtil.getDefaultLocale();
						CollationKey[] collationKeyArray = cache.getCollationKeyArray(sortField.field, locale, indexReader);
						comparableSortArrays.add(collationKeyArray);
					}
				}
				sortIndex++;
			}
		}

		public Comparable[] getDocumentSortComparables(int doc, int distanceSortIndex, CDouble docDistance) {
			Comparable[] documentComparables = new Comparable[sortFields.size()];
			for (int x = 0; x < sortFields.size(); x++) {
				if (x == distanceSortIndex) {
					documentComparables[x] = docDistance;
				} else {
					Comparable[] comparables = comparableSortArrays.get(x);
					documentComparables[x] = comparables[doc];
				}
			}
			return documentComparables;
		}

		public void collect(int doc, float score) {
			if (numHits % 10 == 0) {
				if (!timeoutState.isOK())
					throw new GeneralTimeoutRuntimeException(timeoutState);
			}
			if (filter != null && !filter.exists(doc))
				return;
			if (distance != null) {
				CDouble docDistance = createDistanceCurved(latitudeArray[doc], longitudeArray[doc], distance);
				if (docDistance.value <= distance.distance.value) {
					if (docSet != null)
						docSet.add(doc);
					Comparable[] comparables = getDocumentSortComparables(doc, distanceSortIndex, docDistance);
					CFieldDoc fieldDoc = new CFieldDoc(doc, score, comparables, docDistance, indexReader);
					// if (distanceSortIndex != -1)
					// fieldDoc.fields[distanceSortIndex] = docDistance;
					queue.insert(fieldDoc);
					numHits++;
				}
			} else {
				if (docSet != null)
					docSet.add(doc);
				Comparable[] comparables = getDocumentSortComparables(doc, -1, null);
				CFieldDoc fieldDoc = new CFieldDoc(doc, score, comparables, null, indexReader);
				queue.insert(fieldDoc);
				numHits++;
			}
		}
	}

	public int numDocs(Query a, DocSet b, IndexReader indexReader, TimeoutState timeoutState) throws Exception {
		// Negative query if absolute value different from original
		Query absQ = QueryUtils.getAbs(a);
		DocSet positiveA = cache.getDocSet(absQ, timeoutState, indexReader);
		return a == absQ ? b.intersectionSize(positiveA) : b.andNotSize(positiveA);
	}

	public SearchResult search(Weight weight, Distance distance) throws Exception {
		long numDocsReturned = offset + length;
		CHitQueue queue = new CHitQueue((int) numDocsReturned);
		List<SegmentIndexSearcher> indexSearchers = searchIndexShard.createSegmentIndexSearchers();
		int x = 0;
		CLong numFound = new CLong(0);
		for (SegmentIndexSearcher indexSearcher : indexSearchers) {
			IndexReader indexReader = indexSearcher.getIndexReader();
			DocSet filterDocSet = null;
			if (filterQueries != null) {
				filterDocSet = cache.getDocSet(filterQueries, timeoutState, indexReader);
			}
			ScoreHitCollector hitCollector = new ScoreHitCollector(filterDocSet, queue, indexSearcher, starts[x], distance);
			indexSearcher.search(weight, null, hitCollector);
			processFacets(hitCollector.docSet, indexSearcher);
			numFound.value += hitCollector.numHits;
			x++;
		}
		return new SearchResult(ClusterUtil.toScoreDocs((int) offset, (int) length, queue), numFound, indexSchema);
	}

	public void processFacets(DocSet docSet, SegmentIndexSearcher indexSearcher) throws Exception {
		if (facetResponse != null) {
			IndexReader indexReader = indexSearcher.getIndexReader();
			for (FacetResponse.FieldFacet fieldFacet : facetResponse.fieldFacets) {
				List<Term> terms = ClusterUtil.getFieldTerms(fieldFacet.field, indexReader);
				for (Term term : terms) {
					int count = indexReader.docFreq(term);
					//String readable = ClusterUtil.indexedToReadable(fieldFacet.field, term.text(), indexSchema);
					fieldFacet.add(term.text(), count);
				}
			}
			for (FacetResponse.QueryFacet queryFacet : facetResponse.queryFacets) {
				int count = numDocs(queryFacet.query, docSet, indexReader, timeoutState);
				queryFacet.add(count);
			}
		}
	}
  // TODO: move both facet processings to cache
	public class ScoreHitCollector extends HitCollector {
		DocSet filter;
		public int numHits = 0;
		CHitQueue queue;
		IndexSearcher indexSearcher;
		IndexReader indexReader;
		CLong[] idArray;
		CLong[] versionArray;
		CDouble[] latitudeArray;
		CDouble[] longitudeArray;
		long lastDocRequested;
		float minScore = Float.NEGATIVE_INFINITY;
		int globalDocStart;
		public DocSet docSet;
		Distance distance;

		public ScoreHitCollector(DocSet filter, CHitQueue queue, IndexSearcher indexSearcher, int globalDocStart, Distance distance) throws Exception {
			this.filter = filter;
			this.queue = queue;
			this.indexSearcher = indexSearcher;
			if (facetResponse != null)
				this.docSet = new BitDocSet(indexSearcher.maxDoc());
			this.lastDocRequested = offset + length;
			this.globalDocStart = globalDocStart;
			this.distance = distance;
			this.indexReader = indexSearcher.getIndexReader();
			this.idArray = cache.getLongArray(ClusterConstants.ID_FIELD, indexReader);
			this.versionArray = cache.getLongArray(ClusterConstants.VERSION_FIELD, indexReader);
			if (distance != null) {
				this.latitudeArray = cache.getDoubleArray(ClusterConstants.LATITUDE_FIELD, indexReader);
				this.longitudeArray = cache.getDoubleArray(ClusterConstants.LONGITUDE_FIELD, indexReader);
			}
		}

		// TODO: include distance in document
		public void collect(int doc, float score) {
			if (score > 0.0f) {
				if (numHits % 10 == 0) {
					if (!timeoutState.isOK())
						throw new GeneralTimeoutRuntimeException(timeoutState);
				}
				if (filter != null && !filter.exists(doc))
					return;
				if (numHits++ < lastDocRequested || score >= minScore) {
					CDouble docDistance = null;
					if (distance != null) {
						docDistance = createDistanceCurved(latitudeArray[doc], longitudeArray[doc], distance);
						if (docDistance.value > distance.distance.value) {
							return;
						}
					}
					CScoreDoc scoreDoc = new CScoreDoc(doc, score, docDistance, indexReader);
					queue.insert(scoreDoc);
					if (docSet != null)
						docSet.add(doc);
					minScore = ((CScoreDoc) queue.top()).score;
				}
			}
		}
	}
	/**
	 * public List<Query> rewrite(Query original) throws IOException {
	 * List<Query> rewrittenQueries = new ArrayList<Query>(indexSearchers.size());
	 * for (IndexSearcher indexSearcher : indexSearchers) {
	 * rewrittenQueries.add(indexSearcher.rewrite(original));
	 * }
	 * return rewrittenQueries;
	 * }
	 */
}